package com.atguigu.gmall.realtime.dwd.app;

/*import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.HashMap;
import java.util.Map;

public class DwdBaseLog extends BaseApp {

    private static final String  ERR = "ERR" ;
    private static final String  START = "START" ;
    private static final String  ACTION = "ACTION" ;
    private static final String  DISPLAY = "DISPLAY" ;
    private static final String  PAGE = "PAGE" ;

    public static void main(String[] args) {
        new DwdBaseLog().start(10010, 4, "dwd_base_log", Constant.TOPIC_LOG);

    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 清洗  转换
        SingleOutputStreamOperator<JSONObject> etlStream = etl(stream);
        // etlStream.print("INPUT");


        // 对新老访客标记进行修复
        SingleOutputStreamOperator<JSONObject> fixIsNewDs = fixIsNew(etlStream);
        // fixIsNewDs.print("FIX");


        // 分流 (日志数据格式: 页面日志 + 启动日志)
        Map<String, DataStream<String>> streams = splitStreams(fixIsNewDs);

        // 写出
        writeToKafka(streams);
    }

    private static void writeToKafka(Map<String, DataStream<String>> streams) {
        streams.get(ERR).sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        streams.get(START).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_START));
        streams.get(ACTION).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_ACTION));
        streams.get(DISPLAY).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        streams.get(PAGE).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_PAGE));
    }

    private static Map<String, DataStream<String>> splitStreams(SingleOutputStreamOperator<JSONObject> fixIsNewDs) {
        // 使用侧输出流进行分流操作
        // 错误日志 -> 错误侧输出流 ->  TOPIC_DWD_TRAFFIC_ERR
        // 启动日志 -> 启动侧输出流 ->  TOPIC_DWD_TRAFFIC_START
        // 动作日志 -> 动作侧输出流 ->  TOPIC_DWD_TRAFFIC_ACTION
        // 曝光日志 -> 曝光侧输出流 ->  TOPIC_DWD_TRAFFIC_DISPLAY
        // 页面日志 -> 主流       ->  TOPIC_DWD_TRAFFIC_PAGE

        OutputTag<String> errTag = new OutputTag<>("errTag", Types.STRING);
        OutputTag<String> startTag = new OutputTag<>("startTag", Types.STRING);
        OutputTag<String> actionTag = new OutputTag<>("actionTag", Types.STRING);
        OutputTag<String> displayTag = new OutputTag<>("displayTag", Types.STRING);

        SingleOutputStreamOperator<String> pageDs = fixIsNewDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                        // 错误日志 -> 错误侧输出流 ->  TOPIC_DWD_TRAFFIC_ERR
                        // 判断当前页面日志或者启动日志中是否有err ， 如果有， 直接将整条数据写到对应的侧输出流
                        JSONObject errJsonObj = jsonObj.getJSONObject("err");
                        if (errJsonObj != null) {  // 说明是错误日志
                            context.output(errTag, jsonObj.toJSONString());
                            // 从数据中移除err
                            jsonObj.remove("err");
                        }

                        // 启动日志 -> 启动侧输出流 ->  TOPIC_DWD_TRAFFIC_START
                        JSONObject startJsonObj = jsonObj.getJSONObject("start");
                        if (startJsonObj != null) { // 说明包含启动日志
                            context.output(startTag, jsonObj.toJSONString());
                        }

                        // 页面日志
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");  // 得知道你当前动作属于哪个common
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");  // 得知道你当前动作属于哪个page
                        Long ts = jsonObj.getLong("ts");
                        if (pageJsonObj != null) {
                            // 动作日志 -> 动作侧输出流 ->  TOPIC_DWD_TRAFFIC_ACTION
                            JSONArray actionJsonArr = jsonObj.getJSONArray("actions");
                            if (actionJsonArr != null && actionJsonArr.size() > 0) {
                                for (int i = 0; i < actionJsonArr.size(); i++) {
                                    // 取出每个action
                                    JSONObject actionJsonObj = actionJsonArr.getJSONObject(i);
                                    //将 action 、 page 、 common信息封装 ， 写出
                                    JSONObject newActionJsonObj = new JSONObject();
                                    newActionJsonObj.put("common", commonJsonObj);
                                    newActionJsonObj.put("page", pageJsonObj);
                                    newActionJsonObj.put("action", actionJsonObj);
                                    // 因为action中已经包含了ts, 所以可以不封装外部的ts

                                    // 写出到动作侧输出流
                                    context.output(actionTag, newActionJsonObj.toJSONString());
                                }

                                // 将动作数据移除掉
                                jsonObj.remove("actions");
                            }
                        }


                        // 曝光日志 -> 曝光侧输出流 ->  TOPIC_DWD_TRAFFIC_DISPLAY
                        JSONArray displayJsonArr = jsonObj.getJSONArray("displays");
                        if (displayJsonArr != null && displayJsonArr.size() > 0) {
                            for (int i = 0; i < displayJsonArr.size(); i++) {
                                // 取出每个曝光
                                JSONObject displayJsonObj = displayJsonArr.getJSONObject(i);
                                //将 display 、 page 、 common 、 ts封装
                                JSONObject newDisplayJsonObj = new JSONObject();
                                newDisplayJsonObj.put("common", commonJsonObj);
                                newDisplayJsonObj.put("page", pageJsonObj);
                                newDisplayJsonObj.put("display", displayJsonObj);
                                newDisplayJsonObj.put("ts", ts);

                                // 写出到曝光侧输出流
                                context.output(displayTag, newDisplayJsonObj.toJSONString());
                            }
                            // 移除display
                            jsonObj.remove("display");
                        }


                        // 页面日志 -> 主流       ->  TOPIC_DWD_TRAFFIC_PAGE

                        collector.collect(jsonObj.toJSONString());
                    }
                }
        );
        //捕获侧流
        SideOutputDataStream<String> errDs = pageDs.getSideOutput(errTag);
        SideOutputDataStream<String> startDs = pageDs.getSideOutput(startTag);
        SideOutputDataStream<String> actionDs = pageDs.getSideOutput(actionTag);
        SideOutputDataStream<String> displayDs = pageDs.getSideOutput(displayTag);

        //通过Map封装流
        Map<String, DataStream<String>> streams = new HashMap<>();
        streams.put(ERR , errDs );
        streams.put(START , startDs );
        streams.put(ACTION, actionDs );
        streams.put(DISPLAY , displayDs );
        streams.put(PAGE , pageDs) ;

        pageDs.print("PAGE");
        errDs.print("ERR");
        startDs.print("START");
        actionDs.print("ACTION");
        displayDs.print("DISPLAY");
        return streams;
    }

    private static SingleOutputStreamOperator<JSONObject> fixIsNew(SingleOutputStreamOperator<JSONObject> etlStream) {
        SingleOutputStreamOperator<JSONObject> fixIsNewDs = etlStream.keyBy(
                // 通过 mid 进行keyBy, 针对于每个mid进行标记的修复
                (KeySelector<JSONObject, String>) jsonObject -> jsonObject.getJSONObject("common").getString("mid")
        ).process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    // 声明状态
                    private ValueState<String> lastVisitDataState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastVisitDateStateDesc = new ValueStateDescriptor<>("lastVisitDateStateDesc", Types.STRING);
                        RuntimeContext runtimeContext = getRuntimeContext();
                        lastVisitDataState = runtimeContext.getState(lastVisitDateStateDesc);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        // 取出is_new
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        // 从状态中取出上次的访问日期
                        String lastVisitDate = lastVisitDataState.value();
                        // 当前数据中的日期
                        Long ts = jsonObj.getLong("ts");
                        // 转换成日期 使用自定义工具类DateFormatUtil
                        String currentVisitDate = DateFormatUtil.tsToDate(ts);
                        if (isNew.equals("1")) {  // is_new=1
                            if (lastVisitDate == null) {
                                // 1.状态为null, 说明是新访客, 不进行修复
                                // 更新状态
                                lastVisitDataState.update(currentVisitDate);
                            } else if (!lastVisitDate.equals(currentVisitDate)) {
                                // 2.状态不为null且状态中的日期不是当天, 说明是老访客,需要进行修复
                                jsonObj.getJSONObject("commmon").put("is_new", "0");
                            } else {
                                // 3.状态不为null且状态中的日期是当天, 说明是新访客, 不进行任何处理
                            }
                        } else { // is_new=0
                            // 1.状态为null, 说明访客是数仓上线前就访问过,这次是数仓上线后第一次访问,需要将状态维护成昨天
                            if (lastVisitDate == null) {
                                // 计算昨天
                                String yesterday = DateFormatUtil.tsToDateTime(ts - 1000 * 60 * 60 * 24);
                                lastVisitDataState.update(yesterday);
                            } else {
                                // 2.状态不为null, 不进行任何处理
                            }
                        }

                        // 写出数据
                        collector.collect(jsonObj);
                    }
                }
        );
        return fixIsNewDs;
    }

    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        OutputTag<String> dirtyTag = new OutputTag<>("dirtyTag", Types.STRING);
        SingleOutputStreamOperator<JSONObject> etlStream = stream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(value);
                        } catch (Exception e) {
                            System.out.println("过滤掉脏数据" + value);
                            // 将数据写入到测流
                            context.output(dirtyTag, value);
                        }
                    }
                }
        );
        // 捕获脏数据的测流
        SideOutputDataStream<String> dirtyDs = etlStream.getSideOutput(dirtyTag);
        // 将脏数据写出到Kafka中
        KafkaSink<String> kafkaSink = FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_DIRTY);
        *//*KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(Constant.TOPIC_DWD_DIRTY)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                // DeliveryGuarantee.EXACTLY_ONCE | AT_LEAST_ONCE
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                // 如果是精确一次, 还需要设置 事务id的前缀, 生产者事务超时时间
                .setTransactionalIdPrefix("gmall-realtime-"+System.currentTimeMillis())
                // 默认值: Kafka Broker最大的事务超市时间: 15min
                //          Kafka Sink最大的事务超时时间:  1hour
                // 原则:  检查点超时时间Checkpoint<=Kafka Sink最大的事务超时时间 <= Kafka Broker最大的事务超市时间
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 1000 * 60 + "" )
                .build();*//*

        dirtyDs.sinkTo(kafkaSink);
        return etlStream;
    }
}*/

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerConfig;

import javax.print.attribute.standard.PagesPerMinute;
import javax.swing.*;
import java.util.HashMap;
import java.util.Map;

/**
 * @author WEIYUNHUI
 * @date 2024/8/2 15:30
 */
public class DwdBaseLog extends BaseApp {

    private static final String  ERR = "ERR" ;
    private static final String  START = "START" ;
    private static final String  ACTION = "ACTION" ;
    private static final String  DISPLAY = "DISPLAY" ;
    private static final String  PAGE = "PAGE" ;


    public static void main(String[] args) {
        new DwdBaseLog().start( 10018 , 4 , "dwd_base_log" , Constant.TOPIC_LOG);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 清洗  转换
        SingleOutputStreamOperator<JSONObject> etlStream = etl(stream);
        //etlStream.print("INPUT");

        //对新老访客标记进行修复
        SingleOutputStreamOperator<JSONObject> fixIsNewDs = fixIsNew(etlStream);
        //fixIsNewDs.print("FIX");

        // 分流
        Map<String, DataStream<String>> streams = splitStreams(fixIsNewDs);

        //写出
        writeToKafka(streams);

    }

    private static void writeToKafka(Map<String, DataStream<String>> streams) {

        streams.get(ERR).sinkTo( FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        streams.get(START).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_START));
        streams.get(ACTION).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_ACTION));
        streams.get(DISPLAY).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        streams.get(PAGE).sinkTo( FlinkSinkUtil.getKafkaSink( Constant.TOPIC_DWD_TRAFFIC_PAGE));
    }

    private static Map<String, DataStream<String>> splitStreams(SingleOutputStreamOperator<JSONObject> fixIsNewDs) {
        // 使用侧输出流进行分流操作
        // 错误日志 -> 错误侧输出流 ->  TOPIC_DWD_TRAFFIC_ERR
        // 启动日志 -> 启动侧输出流 ->  TOPIC_DWD_TRAFFIC_START
        // 动作日志 -> 动作侧输出流 ->  TOPIC_DWD_TRAFFIC_ACTION
        // 曝光日志 -> 曝光侧输出流 ->  TOPIC_DWD_TRAFFIC_DISPLAY
        // 页面日志 -> 主流       ->  TOPIC_DWD_TRAFFIC_PAGE

        //侧输出流标签
        OutputTag<String> errTag = new OutputTag<>("errTag", Types.STRING);
        OutputTag<String> startTag = new OutputTag<>("startTag", Types.STRING);
        OutputTag<String> actionTag = new OutputTag<>("actionTag", Types.STRING);
        OutputTag<String> displayTag = new OutputTag<>("displayTag", Types.STRING);

        SingleOutputStreamOperator<String> pageDs = fixIsNewDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                        // 错误日志 -> 错误侧输出流 ->  TOPIC_DWD_TRAFFIC_ERR
                        // 判断当前页面日志或者启动日志中是否有err ， 如果有， 直接将整条数据写到对应的侧输出流
                        JSONObject errJsonObj = jsonObj.getJSONObject("err");
                        if (errJsonObj != null) {
                            ctx.output(errTag, jsonObj.toJSONString());
                            //从数据中将err移除
                            jsonObj.remove("err");
                        }

                        //启动日志 -> 启动侧输出流 ->  TOPIC_DWD_TRAFFIC_START
                        JSONObject startJsonObj = jsonObj.getJSONObject("start");
                        if (startJsonObj != null) {
                            ctx.output(startTag, jsonObj.toJSONString());
                        }

                        // 页面日志
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        Long ts = jsonObj.getLong("ts");
                        if (pageJsonObj != null) {
                            // 动作日志 -> 动作侧输出流 ->  TOPIC_DWD_TRAFFIC_ACTION
                            JSONArray actionJsonArr = jsonObj.getJSONArray("actions");
                            if (actionJsonArr != null && actionJsonArr.size() > 0) {
                                for (int i = 0; i < actionJsonArr.size(); i++) {
                                    //取出每个action
                                    JSONObject actionJsonObj = actionJsonArr.getJSONObject(i);
                                    //将 action 、 page 、 common信息封装 ， 写出
                                    JSONObject newActionJsonObj = new JSONObject();
                                    newActionJsonObj.put("common", commonJsonObj);
                                    newActionJsonObj.put("page", pageJsonObj);
                                    newActionJsonObj.put("action", actionJsonObj);
                                    // 因为action中已经包含了ts, 所以可以不封装外部的ts

                                    //写出到动作侧输出流
                                    ctx.output(actionTag, newActionJsonObj.toJSONString());
                                }

                                //将动作数据移除掉
                                jsonObj.remove("actions");
                            }
                            // 曝光日志 -> 曝光侧输出流 ->  TOPIC_DWD_TRAFFIC_DISPLAY
                            JSONArray displayJsonArr = jsonObj.getJSONArray("displays");
                            if (displayJsonArr != null && displayJsonArr.size() > 0) {
                                for (int i = 0; i < displayJsonArr.size(); i++) {
                                    //取出每个曝光
                                    JSONObject displayJsonObj = displayJsonArr.getJSONObject(i);
                                    //将 display 、 page 、 common 、 ts
                                    JSONObject newDisplayJsonObj = new JSONObject();
                                    newDisplayJsonObj.put("common", commonJsonObj);
                                    newDisplayJsonObj.put("page", pageJsonObj);
                                    newDisplayJsonObj.put("display", displayJsonObj);
                                    newDisplayJsonObj.put("ts", ts);

                                    //写出到曝光侧输出流
                                    ctx.output(displayTag, newDisplayJsonObj.toJSONString());
                                }

                                //移除displays
                                jsonObj.remove("displays");
                            }


                            // 页面日志 -> 主流       ->  TOPIC_DWD_TRAFFIC_PAGE

                            out.collect(jsonObj.toJSONString());
                        }
                    }
                }
        );

        //捕获侧流
        SideOutputDataStream<String> errDs = pageDs.getSideOutput(errTag);
        SideOutputDataStream<String> startDs = pageDs.getSideOutput(startTag);
        SideOutputDataStream<String> actionDs = pageDs.getSideOutput(actionTag);
        SideOutputDataStream<String> displayDs = pageDs.getSideOutput(displayTag);

        //通过Map封装流
        Map<String, DataStream<String>> streams = new HashMap<>();
        streams.put(ERR , errDs );
        streams.put(START , startDs );
        streams.put(ACTION, actionDs );
        streams.put(DISPLAY , displayDs );
        streams.put(PAGE , pageDs) ;

        pageDs.print("PAGE");
        errDs.print("ERR");
        startDs.print("START");
        actionDs.print("ACTION");
        displayDs.print("DISPLAY");
        return streams;
    }

    private static SingleOutputStreamOperator<JSONObject> fixIsNew(SingleOutputStreamOperator<JSONObject> etlStream) {
        SingleOutputStreamOperator<JSONObject> fixIsNewDs = etlStream.keyBy(
                // 通过 mid 进行keyBy, 针对于每个mid进行标记的修复
                jsonObj -> jsonObj.getJSONObject("common").getString("mid")
        ).process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    // 声明状态
                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastVisitDateStateDesc
                                = new ValueStateDescriptor<>("lastVisitDateStateDesc", Types.STRING);

                        lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDesc);

                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        //取出is_new
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        //从状态中取出上次的访问日期
                        String lastVisitDate = lastVisitDateState.value();
                        //当前数据中的日期
                        Long ts = jsonObj.getLong("ts");
                        //转换成日期
                        String currentVisitDate = DateFormatUtil.tsToDate(ts);

                        // is_new = 1
                        if ("1".equals(isNew)) {
                            if (lastVisitDate == null) {
                                // 1. 状态为null, 说明是新访客，不需要进行修复
                                // 更新状态
                                lastVisitDateState.update(currentVisitDate);
                            } else if (!lastVisitDate.equals(currentVisitDate)) {
                                // 2. 状态不为null且状态中的日期不是当天（状态的日期与数据中的日期不一致）， 说明是老访客，需要进行修复
                                jsonObj.getJSONObject("common").put("is_new", "0");
                            } else {
                                // 3. 状态不为null且状态中的日期为当天（状态的日期与数据中的日期一致）， 说明是新访客，不需要进行修复
                            }
                        } else {
                            // is_new = 0
                            // 1. 状态为null ，说明该访客是数仓上线后的首次访问， 需要将状态维护成昨日（今天之前的一个日期即可）
                            if (lastVisitDate == null) {
                                //计算昨日
                                String yesterday = DateFormatUtil.tsToDate(ts - 1000 * 60 * 60 * 24);
                                lastVisitDateState.update(yesterday);
                            } else {
                                // 2. 状态不为null , 不需要进行修复
                            }
                        }

                        //写出数据
                        out.collect(jsonObj);
                    }
                }
        );
        return fixIsNewDs;
    }

    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        OutputTag<String> dirtyTag = new OutputTag<>("dirtyTag", Types.STRING);
        SingleOutputStreamOperator<JSONObject> etlStream = stream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(value);
                            out.collect(jsonObj);
                        } catch (Exception e) {
                            System.out.println("过滤掉脏数据: " + value);
                            //将脏数据写入到侧流
                            ctx.output(dirtyTag , value );
                        }
                    }
                }
        );
        //捕获脏数据的侧流
        SideOutputDataStream<String> dirtyDs = etlStream.getSideOutput(dirtyTag);
        //将脏数据写出到kafka中
        KafkaSink<String> kafkaSink = FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_DIRTY);
        dirtyDs.sinkTo( kafkaSink ) ;
        return etlStream;
    }
}
